package com.atguigu.day4



import com.atguigu.stream.watermark.EventTimeExample.WindowResult
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector


object UpdataWindowResultWithLateElement {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //    设置时间语义为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //    系统默认200毫秒插入一次水位线    这里设为1000毫秒
    env.getConfig.setAutoWatermarkInterval(1000)
    //
    val stream = env.socketTextStream("192.168.35.102", 9999, '\n')

    val str = stream
      .map(line => {
        val arr = line.split(" ")
        (arr(0), arr(1).toLong * 1000L)
      })
      .assignTimestampsAndWatermarks(
        new BoundedOutOfOrdernessTimestampExtractor[(String, Long)](Time.seconds(5)) {
          override def extractTimestamp(t: (String, Long)): Long = t._2
        }
      )
      .keyBy(_._1)
      .timeWindow(Time.seconds  (10))
      //窗口闭合后，等待迟到延迟时间
      .allowedLateness(Time.seconds(5))
      .process(new WindowResult)
      .print()
    env.execute()
  }

    class UpdataWindowResult extends ProcessWindowFunction[(String, Long), String, String, TimeWindow]{

      override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[String]): Unit = {
        //当第一次对窗口求值时，也就是水位线超过窗口结束时间的时候
        //会调用process函数
        //这是isUpdata为默认值false
        //窗口内初始化一个状态变量使用windowState,只对当前窗口可见
        val isUpdata = context.windowState.getState(
          new ValueStateDescriptor[Boolean]("updata", Types.of[Boolean])
        )
        if (!isUpdata.value()) {
          out.collect("窗口第一次进行求值！元素数量共有" + elements.size)
          isUpdata.update(true)

        } else {
          out.collect("迟到元素到来了！更新元素数量为" + elements.size)
        }

      }
    }

}